package ItemMapReduce;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import dataStructures.Ratings;
import dataStructures.TwoVectorsFirstIntSecondDouble;
import dataStructures.UsersMapOutputWritable;

public class ItemProcessReducer extends MapReduceBase implements Reducer<IntWritable, UsersMapOutputWritable, IntWritable, TwoVectorsFirstIntSecondDouble> {


	
	@Override
	public void reduce(IntWritable key, Iterator<UsersMapOutputWritable> val,
			OutputCollector<IntWritable, TwoVectorsFirstIntSecondDouble> out,
			Reporter arg3) throws IOException {
		
		ArrayList <Integer> userIDsList= new ArrayList<Integer>();; 
		ArrayList <Ratings[]> cbElements= new ArrayList<Ratings[]>();
		UsersMapOutputWritable tmp;
		
		int count = 0;
		while(val.hasNext()){
			count++;
			tmp = val.next();
			
			userIDsList = tmp.getUserIdsInIntegers();
			
			cbElements.add(tmp.getCbElements());
		}
		int [] a = toIntArray(userIDsList);
		if(a.length != userIDsList.size()){
			System.out.println("Problems with conversion");
			System.exit(-1);
		}
			
		Ratings [] b = accumulateCbElementsValues(cbElements);
		TwoVectorsFirstIntSecondDouble oVal = new TwoVectorsFirstIntSecondDouble(a , b);
		
		userIDsList.clear();
		cbElements.clear();
		out.collect(key, oVal);
		
	}
	
	
	
	private int [] toIntArray(List <Integer> userIDsList){
		int [] ans = new int [userIDsList.size()];
		
		for(int i  = 0 ; i < userIDsList.size() ; i++){
			ans[i] = userIDsList.get(i).intValue();
		}
		
		return ans;
	}
	
	private Ratings[] accumulateCbElementsValues(List <Ratings[]> cbElements){
		Ratings[] ans = new Ratings[ cbElements.get(0).length];
		for(int i = 0 ; i < ans.length ; i++){
			ans[i] = new Ratings(0,0);
		}
		
		for(Ratings[] r : cbElements){
			if(r.length != ans.length){
				System.out.println("The lengths of the arrays are different");
				System.exit(-1);
			}
			
			for(int i = 0 ; i < ans.length ; i++)
				ans[i].accumulateCbElements(r[i]);
		}
				
		return ans;
	}

	

}
